package com.zyx.pulsardemo.basic.admin;

import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.Topics;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.policies.data.PersistencePolicies;
import org.apache.pulsar.common.policies.data.TenantInfo;

import java.util.HashSet;
import java.util.stream.Collector;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author Yaxi.Zhang
 * @since 2021/9/30 14:46
 * desc: 创建pulsar主题
 * reference: https://github.com/david-streamlio/pulsar-in-action
 */
public class CreateTopic {
    public static void main(String[] args) throws Exception {
        // 需要 pulsar-client-admin 依赖
        PulsarAdmin admin = PulsarAdmin.builder()
                .serviceHttpUrl("http://bigdata022:8080,bigdata023:8080,bigdata024:8080")
                .build();

        // 列出pulsar实例中的 所有集群
        System.out.println("List of clusters");
        admin.clusters().getClusters().forEach(System.out::println);

        System.out.println("====================");

        // 列出pulsar实例中的所 有租户
        System.out.println("List of tenants");
        admin.tenants().getTenants().forEach(System.out::println);

        System.out.println("====================");

        TenantInfo config = TenantInfo.builder()
                // 为租户指定admin角色
                .adminRoles(Stream.of("admin").collect(Collectors.toCollection(HashSet::new)))
                // 指定租户可以操作的集群
                .allowedClusters(Stream.of("eth-pulsar-cluster").collect(Collectors.toCollection(HashSet::new)))
                .build();

        // 创建租户
        /*admin.tenants().createTenant("manning", config);*/

        // 列出 pulsar 中的所有租户
        System.out.println("List of tenants");
        admin.tenants().getTenants().forEach(System.out::println);
        System.out.println("====================");

        // 创建命名空间
        /*admin.namespaces().createNamespace("manning/chapter03");*/

        // 可以对namespace指定 持久化 策略:
        //      int bookkeeperEnsemble: 单个 topic 使用的 bookie 数量, 默认值：0, 该参数类似kafka中的副本数
        //      int bookkeeperWriteQuorum: 每个 entry 要写入的次数, 默认值：0
        //      int bookkeeperAckQuorum: 每个 entry 在等待的 acks（有保证的副本）数量, 默认值：0
        //      double managedLedgerMaxMarkDeleteRate: 标记-删除操作的限制速率（0表示无限制）, 默认值：0.0
        // admin.namespaces().setPersistence("manning/chapter03", new PersistencePolicies(
        //         2, 2, 3, 0.0
        // ));

        // 列出指定租户的所有 命名空间
        System.out.println("List of namespaces under manning tenant");
        admin.namespaces().getNamespaces("manning").forEach(System.out::println);
        System.out.println("====================");

        // 创建不带分区的topic
        admin.topics().createNonPartitionedTopic("persistent://manning/chapter03/example-topic");
        // 创建带有分区的topic
        // admin.topics().createPartitionedTopic("persistent://manning/chapter03/example-topic", 4);
        Topics topics = admin.topics();

        // 创建主题时也可以指定持久化策略, 参数与创建namespaces的类似
        // admin.topics().setPersistence("persistent://manning/chapter03/example-topic", new PersistencePolicies(
        //         2, 2, 3, 0.0));


        // 列出指令命名空间下的所有主题
        System.out.println("List of topics in the manning/chapter03 namespace");
        admin.topics().getList("manning/chapter03").forEach(System.out::println);

        // 关闭admin
        admin.close();
    }
}
